package com.henry.register;

import com.henry.config.CustomerRocketMQProperties;
import com.henry.consumer.ConcurrentConsumerHandler;
import com.henry.consumer.OrderlyConsumerHandler;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;

/**
 * 全局消费者注册器
 *
 * @author Liu Hanlin
 * @create 2025-11-10 23:30
 */
@Slf4j
@Component
public class GlobalConsumerRegister {

    @Resource
    private CustomerRocketMQProperties customerRocketMQProperties;

    @Autowired(required = false)
    private List<ConcurrentConsumerHandler> concurrentConsumers;

    @Autowired(required = false)
    private List<OrderlyConsumerHandler> orderlyConsumers;

    private final String GROUP_TEMPLATE = "%s_%s";

    private final List<DefaultMQPushConsumer> consumerList = new ArrayList<>();

    /**
     * spring bean的声明周期 创建示例、属性注入、初始化、使用、销毁
     * 注册消费者 (在 GlobalConsumerRegister 注入完属性之后进行初始化)
     */
    @PostConstruct
    public void registerConsumer() {

        if (concurrentConsumers != null && !concurrentConsumers.isEmpty()) {
            // 并发消费者注册
            concurrentConsumers.forEach(consumer -> {
                try {
                    DefaultMQPushConsumer pushConsumer = buildConcurrentConsumer(consumer);
                    pushConsumer.start();
                    consumerList.add(pushConsumer);
                    log.info("并发消费者注册成功, topic:[{}], consumerType:[{}]", consumer.getTopic(), consumer.getClass().getName());
                } catch (Exception e) {
                    log.error("并发消费者注册失败, topic:[{}], consumerType:[{}]", consumer.getTopic(), consumer.getClass().getName(), e);
                    throw new RuntimeException("消费者注册失败");
                }
            });
        }

        if (orderlyConsumers != null && !orderlyConsumers.isEmpty()) {
            // 顺序消费者注册
            orderlyConsumers.forEach(consumer -> {
                try {
                    DefaultMQPushConsumer pushConsumer = buildOrderlyConsumer(consumer);
                    pushConsumer.start();
                    consumerList.add(pushConsumer);
                    log.info("顺序消费者注册成功, topic:[{}], consumerType:[{}]", consumer.getTopic(), consumer.getClass().getName());
                } catch (Exception e) {
                    log.error("顺序消费者注册失败, topic:[{}], consumerType:[{}]", consumer.getTopic(), consumer.getClass().getName(), e);
                }
            });
        }



    }

    /**
     * 创建 rocketmq 并发消费者
     * @param concurrentConsumer
     * @return
     */
    private DefaultMQPushConsumer buildConcurrentConsumer(ConcurrentConsumerHandler concurrentConsumer) throws MQClientException {

        // 组名统一使用应用组前缀 + 订阅的主题名
        String groupName = String.format(GROUP_TEMPLATE, customerRocketMQProperties.getConsumerGroupPrefix(), concurrentConsumer.getTopic());
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
        consumer.setNamesrvAddr(customerRocketMQProperties.getNameServer());
        consumer.setConsumeThreadMin(concurrentConsumer.getMinConsumeThreads());
        consumer.setConsumeThreadMax(concurrentConsumer.getMaxConsumeThreads());
        consumer.setMessageModel(concurrentConsumer.getConsumerMode());
        consumer.setMessageListener(concurrentConsumer);
        consumer.subscribe(concurrentConsumer.getTopic(), concurrentConsumer.getTag());
        // 设置消费者首次启动时的初始消费位置
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        return consumer;

    }

    /**
     * 创建 rocketmq 顺序消息消费者
     * @param orderlyConsumer
     * @return
     */
    private DefaultMQPushConsumer buildOrderlyConsumer(OrderlyConsumerHandler orderlyConsumer) throws MQClientException {
        String groupName = String.format(GROUP_TEMPLATE, customerRocketMQProperties.getConsumerGroupPrefix(), orderlyConsumer.getTopic());
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
        consumer.setNamesrvAddr(customerRocketMQProperties.getNameServer());
        consumer.setMessageModel(orderlyConsumer.getConsumerMode());
        consumer.setMessageListener(orderlyConsumer);
        consumer.subscribe(orderlyConsumer.getTopic(), orderlyConsumer.getTag());
        // 设置消费者首次启动时的初始消费位置
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        return consumer;
    }

    @PreDestroy
    private void destroy() {
        consumerList.forEach(consumer -> {
            try {
                consumer.shutdown();
            } catch (Exception e) {
                log.error("消费者销毁失败...", e);
            }
        });
    }


}
